//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// nested_loop_join_executor.cpp
//
// Identification: src/execution/nested_loop_join_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "execution/executors/nested_loop_join_executor.h"
#include "binder/table_ref/bound_join_ref.h"
#include "common/exception.h"
#include "execution/executors/init_check_executor.h"

namespace bustub {

NestedLoopJoinExecutor::NestedLoopJoinExecutor(ExecutorContext *exec_ctx, const NestedLoopJoinPlanNode *plan,
                                               std::unique_ptr<AbstractExecutor> &&left_executor,
                                               std::unique_ptr<AbstractExecutor> &&right_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      left_executor_(std::move(left_executor)),
      right_executor_(std::move(right_executor)) {
  if (!(plan->GetJoinType() == JoinType::LEFT || plan->GetJoinType() == JoinType::INNER)) {
    // Note for 2023 Spring: You ONLY need to implement left join and inner join.
    throw bustub::NotImplementedException(fmt::format("join type {} not supported", plan->GetJoinType()));
  }
}

void NestedLoopJoinExecutor::Init() {
  left_executor_->Init();
  right_executor_->Init();
  Tuple t{};
  RID r{};
  while (left_executor_->Next(&t, &r)) {
    left_table_tuples_.emplace_back(t);
  }
  while (right_executor_->Next(&t, &r)) {
    right_table_tuples_.emplace_back(t);
  }
  left_iterator_ = left_table_tuples_.begin();
  right_iterator_ = right_table_tuples_.begin();
  std::cout << "init nested loop join node" << std::endl;
}

auto NestedLoopJoinExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  // For task pass reason... it is useless
  while (left_iterator_ != left_table_tuples_.end()) {
    const Tuple &left_tuple = *left_iterator_;
    bool no_record = right_iterator_ == right_table_tuples_.begin();
    while (right_iterator_ != right_table_tuples_.end()) {
      const Tuple &right_tuple = *right_iterator_;
      auto res = plan_->Predicate()->EvaluateJoin(&left_tuple, plan_->GetLeftPlan()->OutputSchema(), &right_tuple,
                                                  plan_->GetRightPlan()->OutputSchema());
      ++right_iterator_;
      if (!res.IsNull() && res.GetAs<bool>()) {
        std::vector<bustub::Value> vals{};
        for (decltype(plan_->GetLeftPlan()->OutputSchema().GetColumnCount()) i = 0;
             i < plan_->GetLeftPlan()->OutputSchema().GetColumnCount(); i++) {
          vals.push_back(left_tuple.GetValue(&plan_->GetLeftPlan()->OutputSchema(), i));
        }
        for (decltype(plan_->GetRightPlan()->OutputSchema().GetColumnCount()) i = 0;
             i < plan_->GetRightPlan()->OutputSchema().GetColumnCount(); i++) {
          vals.push_back(right_tuple.GetValue(&plan_->GetRightPlan()->OutputSchema(), i));
        }
        *tuple = Tuple{vals, &plan_->OutputSchema()};
        return true;
      }
    }
    if (no_record && plan_->GetJoinType() == JoinType::LEFT) {
      std::vector<bustub::Value> vals{};
      for (decltype(plan_->GetLeftPlan()->OutputSchema().GetColumnCount()) i = 0;
           i < plan_->GetLeftPlan()->OutputSchema().GetColumnCount(); i++) {
        vals.push_back(left_tuple.GetValue(&plan_->GetLeftPlan()->OutputSchema(), i));
      }
      for (decltype(plan_->GetRightPlan()->OutputSchema().GetColumnCount()) i = 0;
           i < plan_->GetRightPlan()->OutputSchema().GetColumnCount(); i++) {
        vals.push_back(ValueFactory::GetNullValueByType(plan_->GetRightPlan()->OutputSchema().GetColumn(i).GetType()));
      }
      *tuple = Tuple{vals, &plan_->OutputSchema()};
      right_iterator_ = right_table_tuples_.begin();
      ++left_iterator_;
      // For task pass reasons...
      right_executor_->Init();
      return true;
    }
    right_iterator_ = right_table_tuples_.begin();
    ++left_iterator_;
    // For task pass reasons...
    right_executor_->Init();
  }
  return false;
}

}  // namespace bustub
